package com.chinasoft.shop

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import java.sql.DriverManager

object Top20Rest {
  def main(args: Array[String]): Unit = {
    // 初始化 SparkSession
    val spark = SparkSession.builder()
      .appName("Top20Rest")
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._

    // JDBC 配置
    val url = "jdbc:mysql://localhost:3306/dazhong?useSSL=false&characterEncoding=utf8&serverTimezone=UTC"
    val props = new java.util.Properties()
    props.setProperty("user", "root")
    props.setProperty("password", "Etestnmm4l!")
    props.setProperty("driver", "com.mysql.cj.jdbc.Driver")

    try {
      // 读取点评表
      println("开始读取点评数据...")
      val reviewDF = spark.read
        .jdbc(url, "dazhong_dianping", props)

      println(s"成功读取 ${reviewDF.count()} 条点评数据")

      // 计算加权评分
      println("正在计算加权评分...")
      val weightedDF = reviewDF.withColumn(
        "weighted_rating",
        col("rating") * 0.4 +
          col("rating_env") * 0.2 +
          col("rating_flavor") * 0.2 +
          col("rating_service") * 0.2
      )

      // 计算所有饭店的平均加权评分和评论数
      println("正在计算餐厅平均分和评论数...")
      val avgDF = weightedDF.groupBy("restId", "name")
        .agg(
          avg("weighted_rating").alias("average_rating"),
          count("*").alias("review_count")
        )
        .filter("review_count >= 10")  // 过滤掉评论数少于10条的餐厅
        .orderBy(desc("average_rating"))

      // 写入 dazhong_average 表（覆盖写入所有饭店）
      println(s"准备写入结果表，共 ${avgDF.count()} 家餐厅")
      avgDF.write
        .mode("overwrite")
        .jdbc(url, "dazhong_average", props)

      println("结果表写入完成")

      // 打印平均评分最高的前 20 家饭店
      println("===== Top 20 Restaurants by Average Rating =====")
      val top20DF = avgDF.limit(20)
      top20DF.show(20, truncate = false)

      // 可选：将Top20写入单独的表
      top20DF.write
        .mode("overwrite")
        .jdbc(url, "dazhong_top20", props)

      println("Top20餐厅已单独保存到dazhong_top20表")

    } catch {
      case e: Exception =>
        println(s"执行过程中发生错误: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保SparkSession关闭
      spark.stop()
      println("SparkSession已关闭")
    }
  }
}